package com.muyu.danmo.mq;


import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.extra.pinyin.PinyinUtil;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.protobuf.ExtensionRegistry;
import com.muyu.danmo.domain.DanmoInformationPositions;
import com.muyu.danmo.domain.DanmoPersonInfo;
import com.muyu.danmo.domain.DanmoPersonPosition;
import com.muyu.danmo.domain.dto.CanalFlatMessageDTO;
import com.muyu.danmo.domain.es.PersonInfoEs;
import com.muyu.danmo.domain.vo.PositionInfoVO;
import com.muyu.danmo.esMapper.PersonInfoEsRepository;
import com.muyu.danmo.mapper.DanmoPersonInfoMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.elasticsearch.core.ElasticsearchOperations;
import org.springframework.data.elasticsearch.core.query.Query;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;


/**
 * @author Administrator
 */
@Slf4j
@Component
@RocketMQMessageListener(
        topic = "canal-topic",
        consumerGroup = "canal-group",
        selectorExpression = "*"
)
public class BinlogMQMessageListener implements RocketMQListener<String> {

    @Autowired
    private PersonInfoEsRepository esRepo;

    @Autowired
    private DanmoPersonInfoMapper danmoPersonInfoMapper;

    @Autowired
    private ElasticsearchOperations esOps;

    public static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();



    @Override
    public void onMessage(String messageExt) {
        log.info("binlog_info:{}", messageExt);
        CanalFlatMessageDTO msg = null;
        try {
            msg = OBJECT_MAPPER.readValue(messageExt, CanalFlatMessageDTO.class);
        } catch (JsonProcessingException e) {
            throw new RuntimeException(e);
        }
        String type = msg.getType();
        if(StrUtil.isEmpty(type)){
            return;
        }
        if (!List.of("INSERT", "UPDATE", "DELETE").contains(type)) {
            return;
        }

        List<Map<String, Object>> data = msg.getData();
        String table = msg.getTable();
        for (int i = 0; i < data.size(); i++) {
            Map<String, Object> map = data.get(i);
            if ("DELETE".equals(type)) {
                switch (table){
                    case "danmo_person_info_0", "danmo_person_info_2", "danmo_person_info_1":
                        esRepo.deleteById(Long.valueOf((String) map.get("id")));
                        break;
                    case "danmo_person_info_position":
                        esRepo.deleteById(Long.valueOf((String) map.get("person_id")));
                        break;
                    default:
                        log.info("未匹配到表名1：{}", table);
                }
            }else {
                List<PersonInfoEs> list = new ArrayList<>();
                switch (table){
                    case "danmo_person_info_0", "danmo_person_info_2", "danmo_person_info_1":
                        list = danmoPersonInfoMapper.findAllById(Long.valueOf((String) map.get("id")));
                        break;
                    case "danmo_person_position":
                        list = danmoPersonInfoMapper.findAllById(Long.valueOf((String) map.get("person_id")));
                        break;
                    default:
                        log.info("未匹配到表名2：{}", table);
                        continue;
                }
                if(!list.isEmpty()){
                    list.forEach(a ->{
                        if(StrUtil.isEmpty(a.getName())){
                            a.setFirstLetter("#");
                        }else {
                            a.setFirstLetter(String.valueOf(PinyinUtil.getFirstLetter(a.getName().charAt(0))));
                        }
                        if(StrUtil.isEmpty(a.getAddressEntry())){
                            a.setEntryFirstLetter("#");
                        }else {
                            a.setEntryFirstLetter(String.valueOf(PinyinUtil.getFirstLetter(a.getAddressEntry().charAt(0))));
                        }
                    });
                    esRepo.saveAll(list);
                    log.info("数据更新到es完成！！");
                }

            }
        }
    }


}